package com.seu.me.zookeeper.curator.barrier;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
 * DistributedBarrier类实现了栏栅的功能：
 * public DistributedBarrier(CuratorFramework client, String barrierPath)
 * 首先需要调用setBarrier()方法设置栏栅，它将阻塞在它上面等待的线程，
 * 然后需要阻塞的线程调用waitOnBarrier()方法等待放行条件。
 * 当条件满足时调用removeBarrier()方法移除栏栅，所有等待的线程将继续执行。
 *
 * @author liangfeihu
 * @since 2017/12/5 21:16.
 */
public class CuratorBarrier2 {

    /**
     * zookeeper地址
     */
    static final String CONNECT_ADDR = "192.168.169.129:2181,192.168.169.130:2181,192.168.169.131:2181";
    /**
     * session超时时间 ms
     */
    static final int SESSION_OUTTIME = 5000;

    static DistributedBarrier barrier = null;

    public static void main(String[] args) throws Exception {

        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
                        CuratorFramework cf = CuratorFrameworkFactory.builder()
                                .connectString(CONNECT_ADDR)
                                .sessionTimeoutMs(SESSION_OUTTIME)
                                .retryPolicy(retryPolicy)
                                .build();
                        cf.start();
                        barrier = new DistributedBarrier(cf, "/super3");
                        System.out.println(Thread.currentThread().getName() + "设置barrier!");
                        barrier.setBarrier();        //设置
                        barrier.waitOnBarrier();    //等待
                        System.out.println(Thread.currentThread().getName() + "---------开始执行程序----------");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "t" + i).start();
        }

        Thread.sleep(5000);
        barrier.removeBarrier();    //释放

    }
}
